package edu.utexas.ipc;

import java.util.HashMap;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

class Multiplexer implements MessageListener {

	private Clock clock;
	private Linker linker;
	private Map<String, BlockingQueue<Message>> queueMap = new HashMap<String, BlockingQueue<Message>>();
	
	Multiplexer(Clock clock, Linker linker) {
		this.clock = clock;
		this.linker = linker;
	}
	
	public synchronized void messageReceived(Message msg) {

		// update the clock with the incoming timestamp
		clock.update(msg.sourceID, msg.timestamp);

		//System.out.println("multiplexer.recv: " + msg.toString());
		
		// multiplexers talk to each other using this tag
		if (msg.tag.equals("multiplexer")) {
			processCommand(msg);
			return;
		}
		
		// forward the message to the appropriate queue.
		BlockingQueue<Message> queue;
		if((queue = queueMap.get(msg.tag)) != null) {
			queue.add(msg);
		}
		else {
			// System.out.println("message lost: " + msg.toString());
			
			// all mutex connection tags start with this string.
			// ACK any unhandled mutex requests.  This is intended to
			// handle closed mutex connections.
			if (msg.tag.startsWith("__mutex__") && msg.appData.startsWith("request")) {
				sendMessage(msg.sourceID, msg.tag, "nack " + clock.value());
			}
		}
	}

	synchronized Connection openConnection(String tag) {
		BlockingQueue<Message> queue = queueMap.get(tag);
		if (queue == null) {
			queue = new ArrayBlockingQueue<Message>(10);
			queueMap.put(tag, queue);
		}
		broadcast("multiplexer", "open " + tag);
		return new IPCConnection(tag, queue, this);
	}
	
	synchronized void closeConnection(String tag) {
		queueMap.remove(tag);
	}
	
	synchronized void sendMessage(int destID, String tag, String message) {
		
		// System.out.println("multiplexer.send " + destID + " " + tag + " " + clock.value() + " " + message);
		
		linker.sendMessage(destID, tag, clock.value(), message);
		clock.tick();
	}
	
	synchronized void broadcast(String tag, String message) {
		
		// System.out.println("multiplexer.broadcast " + " " + tag + " " + clock.value() + " " + message);
		
		linker.broadcast(tag, clock.value(), message);
		clock.tick();
	}

	synchronized boolean isConnected(int processID) {
		return linker.isConnected(processID);
	}

	public synchronized void connectionLost(int processID) {
		Message m = new Message(processID, 0, "connection_lost", "0", "");
		for(BlockingQueue<Message> q: queueMap.values()) {
			q.add(m);
		}
	}
	
	private void processCommand(Message msg) {
		StringTokenizer st = new StringTokenizer(msg.appData);
		String cmd = st.nextToken();
		if (cmd.equals("open")) {
			String tag = st.nextToken();
			if (!queueMap.containsKey(tag)) {
				BlockingQueue<Message> queue = new ArrayBlockingQueue<Message>(10);
				queueMap.put(tag, queue);
			}
		}	
	}
}
